import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

public class JoinwithDataSetSizeHint {

        public static void main(String[] args) throws Exception {

    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    String rootPath = System.getProperty("user.dir");

    DataSet<Tuple2<Integer, String>> input1 = env.readCsvFile("file://"+rootPath+"/"+"group.csv").types(Integer.class, String.class);
    DataSet<Tuple2<Integer, String>> input2 = env.readCsvFile("file://"+rootPath+"/"+"group2.csv").types(Integer.class, String.class);


            System.out.println("--------------------input1------------------------------------");
            input1.print();
            System.out.println("--------------------input2------------------------------------");
            input2.print();

//        如果两套数据集中的各自一行的key0相同,那么就拼接这两个数据
    DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<Integer, String>>>
            result1 =
            // hint that the second DataSet is very small
            input1.joinWithTiny(input2)
                    .where(0)
                    .equalTo(0);
    System.out.println("------------------result1--------------------------------------");
    result1.print();
//
    System.out.println("------------------result2--------------------------------------");
    DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<Integer, String>>>
            result2 =
            // hint that the second DataSet is very large
            input1.joinWithHuge(input2)
                    .where(0)
                    .equalTo(0);


        result2.print();

     }

}
